%%%====================================================================
%%% File    : pipe_join.erl
%%% Author  :  <vjache>
%%% Description: This is a simpliest pipe implementation. It may be 
%%%				 considered as a simple reference implementation of 
%%%				 a pipe callback module, and could be used for debug 
%%%				 purposes.
%%%
%%% Created : 30.03.2009
%%%====================================================================

-module(pipe_join).

%%
%% Include files
%%
-include("../include/pipe.hrl").
%%
%% Exported Functions
%%
-export([
		 new/2%,test/1
		]).

%%====================================================================
%% API Functions
%%====================================================================

new(JoinSpecs,NextPipe) ->
	pipe_folder:new(folder(JoinSpecs), NextPipe).

%%====================================================================
%% Test Functions
%%====================================================================

%% test(L) ->
%% 	S=pipe:open(new()),
%% 	{IterState,PipeState}=pipe:push_each(S, iter:from_list(L)),
%% 	iter:close(IterState),
%% 	pipe:close(PipeState).


%%====================================================================
%% Local Functions
%%====================================================================

folder(JoinSpecs) ->
	#folder{
			open_fun=fun()-> 
							 {_KeysTid=ets:new(keys, []),dict:new()}
					 end,%% Returns a new initial folder
			fold_fun=fun fold_fun/2,%% Folds Item with AccIn and retuns new one
			close_fun=fun({KeysTid,KvTidsBySign})-> 
							  true=ets:delete(KeysTid),
							  dict:fold(fun(_SS,Tid,_)->
												true=ets:delete(Tid)
										end, [], KvTidsBySign),
							  ok 
					  end,%% Releases resources of folder if any
			result_fun=fun({KeysTid,KvTidsBySign})-> 
							   TidKfList=align_tids(JoinSpecs,KvTidsBySign),
							   case TidKfList of
								   incomplete ->
									   [];
								   _ ->
									   LookupFuns=[fun(Key)->
														iter:from_list(ets:lookup(KvTid, Key))
												end||{KvTid,_KeyFun}<-TidKfList],
									   iter_join_lookup:new_flat_ls(iter_ets_keys:new(KeysTid),LookupFuns)
							   end
					   end,%% Converts accumulator to result before pass to next pipe
			capacity=-1 %% Returns folder capacity (-1 denotes Infinity). If exhausted then accumulated value pass to next pipe and folder destroyed.
		   }.

align_tids([],_KvTidsBySign) ->
	[];
align_tids([{Sign,KeyFun}|JoinSpecsRest],KvTidsBySign) ->
	Res=dict:find(Sign, KvTidsBySign),
	case Res of
		error ->
			incomplete;
		{ok,Tid} ->
			case align_tids(JoinSpecsRest,KvTidsBySign) of
				incomplete ->
					incomplete;
				R ->
					[{Tid,KeyFun}|R]
			end
	end.


fold_fun({SenderSign,RecordList}=_Item,StateIn) when is_list(RecordList)->
	lists:foldl(
	  fun(Record,AccIn)->
			  fold_fun({SenderSign,Record},AccIn)
	  end, StateIn, RecordList);
fold_fun(?EOF=_Item,AccIn)->
	AccIn;
fold_fun({SenderSign,Record}=_Item,{KeysTid,KvTidsBySign}=_AccIn)->
	{NewKvTidsBySign,KvTid}=case dict:find(SenderSign, KvTidsBySign) of
								error ->
									Tid=ets:new(records, [bag,{keypos,2}]),
									{dict:store(SenderSign, Tid, KvTidsBySign),Tid};
								{ok,Tid} ->
									{KvTidsBySign,Tid}
							end,
 	Key=element(2,Record),
%% 	Value=Record,
%% 	true=ets:insert(KvTid, {Key,Value}), 
	true=ets:insert(KvTid, Record), 
	true=ets:insert(KeysTid, {Key}),
	{KeysTid,NewKvTidsBySign};
fold_fun(F,_S) ->
	throw({expected_ss_format,F}).
